﻿/*  ----------------------------------------------------------------------------
 *  SysServer
 *  ----------------------------------------------------------------------------
 *  File:       IOCPThreadPool.cs
 *  Creater:     SEEN\MistSeen
 *  ----------------------------------------------------------------------------
 *  $Date: 2010-09-06 10:36:01 +0000 (Mon, 06 Sep 2010) $
 *  $Rev: 15 $
 *  $Author: ScorDeity $
 *  $URL: http://sysserver.googlecode.com/svn/trunk/src/SysServer/Threading/IOCPThreadPool.cs $
 *  $Id: IOCPThreadPool.cs 15 2010-09-06 10:36:01Z ScorDeity $
 */

namespace SysServer.Threading
{
    using System;
    using System.Threading;

    /// <summary>
    /// this class provides the ability to create a thread pool to manage work.  The
    ///           class abstracts the Win32 IOCompletionPort API so it requires the use of
    ///           unmanaged code.  Unfortunately the .NET framework does not provide this functionality
    /// </summary>
    public sealed class IOCPThreadPool
    {
        // Delegate Function Types
        /// <summary> DelType: This is the type of user function to be supplied for the thread pool </summary>
        public delegate void FuncHandle(Int32 iValue);



        // Private Properties
        private UInt32 _Handle;
        /// <summary> SimType: Contains the IO Completion Port Thread Pool handle for this instance </summary>
        private UInt32 GetHandle { get { return _Handle; } set { _Handle = value; } }

        private Int32 m_MaxConcurrency;
        /// <summary> SimType: The maximum number of threads that may be running at the same time </summary>
        private Int32 MaxConcurrency { get { return m_MaxConcurrency; } set { m_MaxConcurrency = value; } }

        private Int32 m_iMinThreadsInPool;
        /// <summary> SimType: The minimal number of threads the thread pool maintains </summary>
        private Int32 MinThreadsInPool { get { return m_iMinThreadsInPool; } set { m_iMinThreadsInPool = value; } }

        private Int32 m_iMaxThreadsInPool;
        /// <summary> SimType: The maximum number of threads the thread pool maintains </summary>
        private Int32 MaxThreadsInPool { get { return m_iMaxThreadsInPool; } set { m_iMaxThreadsInPool = value; } }

        private Object m_pCriticalSection;
        /// <summary> RefType: A serialization object to protect the class state </summary>
        private Object CriticalSection { get { return m_pCriticalSection; } set { m_pCriticalSection = value; } }

        private FuncHandle m_pfnUserFunction;
        /// <summary> DelType: A reference to a user specified function to be call by the thread pool </summary>
        private FuncHandle UserFunction { get { return m_pfnUserFunction; } set { m_pfnUserFunction = value; } }

        private Boolean m_bDisposeFlag;
        /// <summary> SimType: Flag to indicate if the class is disposing </summary>
        private Boolean IsDisposed { get { return m_bDisposeFlag; } set { m_bDisposeFlag = value; } }



        private Int32 m_ThreadsInPool;
        /// <summary> SimType: The current number of threads in the thread pool </summary>
        public Int32 ThreadsInPool { get { return m_ThreadsInPool; } set { m_ThreadsInPool = value; } }
        /// <summary> SimType: Increment current number of threads in the thread pool </summary>
        private Int32 IncCurrentThreadsInPool() { return Interlocked.Increment(ref m_ThreadsInPool); }
        /// <summary> SimType: Decrement current number of threads in the thread pool </summary>
        private Int32 DecCurrentThreadsInPool() { return Interlocked.Decrement(ref m_ThreadsInPool); }
        private Int32 m_ActiveThreadsInPool;
        /// <summary> SimType: The current number of active threads in the thread pool </summary>
        public Int32 ActiveThreadsInPool { get { return m_ActiveThreadsInPool; } set { m_ActiveThreadsInPool = value; } }
        /// <summary> SimType: Increment current number of active threads in the thread pool </summary>
        private Int32 IncActThreadsInPool() { return Interlocked.Increment(ref m_ActiveThreadsInPool); }
        /// <summary> SimType: Decrement current number of active threads in the thread pool </summary>
        private Int32 DecActThreadsInPool() { return Interlocked.Decrement(ref m_ActiveThreadsInPool); }
        private Int32 m_CurrentWorkInPool;
        /// <summary> SimType: The current number of Work posted in the thread pool </summary>
        public Int32 WorkInPool { get { return m_CurrentWorkInPool; } set { m_CurrentWorkInPool = value; } }
        /// <summary> SimType: Increment current number of Work posted in the thread pool </summary>
        private Int32 IncCurrentWorkInPool() { return Interlocked.Increment(ref m_CurrentWorkInPool); }
        /// <summary> SimType: Decrement current number of Work posted in the thread pool </summary>
        private Int32 DecCurrentWorkInPool() { return Interlocked.Decrement(ref m_CurrentWorkInPool); }




        /// <summary> Constructor </summary>
        /// <param name = "iMaxConcurrency"> SimType: Max number of running threads allowed </param>
        /// <param name = "iMinThreadsInPool"> SimType: Min number of threads in the pool </param>
        /// <param name = "iMaxThreadsInPool"> SimType: Max number of threads in the pool </param>
        /// <param name = "func"> DelType: Reference to a function to call to perform work </param>
        /// <exception cref = "Exception"> Unhandled Exception </exception>
        public IOCPThreadPool(Int32 iMaxConcurrency, Int32 iMinThreadsInPool, Int32 iMaxThreadsInPool, FuncHandle func)
        {
            try
            {
                // Set initial class state
                this.MaxConcurrency = iMaxConcurrency;
                this.MinThreadsInPool = iMinThreadsInPool;
                this.MaxThreadsInPool = iMaxThreadsInPool;
                this.UserFunction = func;
                // Init the thread counters
                this.ThreadsInPool = 0;
                this.ActiveThreadsInPool = 0;
                this.WorkInPool = 0;
                // Initialize the Monitor Object
                this.CriticalSection = new Object();
                // Set the disposing flag to false
                this.IsDisposed = false;
                unsafe
                {
                    // Create an IO Completion Port for Thread Pool use
                    GetHandle = IOCPEx.CreateIoCompletionPort(IOCPEx.INVALID_HANDLE_VALUE, 0, null, (UInt32)MaxConcurrency);
                }
                // Test to make sure the IO Completion Port was created
                if (GetHandle == 0)
                {
                    throw new Exception("Unable To Create IO Completion Port");
                }
                // Allocate and start the Minimum number of threads specified
                Int32 iStartingCount = ThreadsInPool;
                ThreadStart tsThread = new ThreadStart(IOCPFunction);

                for (Int32 iThread = 0; iThread < MinThreadsInPool; ++iThread)
                {
                    // Create a thread and start it
                    Thread thThread = new Thread(tsThread);
                    thThread.Name = "IOCP " + thThread.GetHashCode();
                    thThread.Start();
                    // Increment the thread pool count
                    IncCurrentThreadsInPool();
                }
            }
            catch
            {
                throw new Exception("Unhandled Exception");
            }
        }

        /// <summary> Finalize called by the GC </summary>
        ~IOCPThreadPool()
        {
            if (!IsDisposed)
                Dispose();
        }
        /// <summary> Called when the object will be shutdown.  This
        ///           function will wait for all of the work to be completed
        ///           inside the queue before completing </summary>
        public void Dispose()
        {
            try
            {
                // Flag that we are disposing this object
                IsDisposed = true;
                // Get the current number of threads in the pool
                Int32 iCurThreadsInPool = ThreadsInPool;
                // Shutdown all thread in the pool
                for (Int32 iThread = 0; iThread < iCurThreadsInPool; ++iThread)
                {
                    unsafe
                    {
                        bool bret = IOCPEx.PostQueuedCompletionStatus(GetHandle, 4, (UInt32*)IOCPEx.SHUTDOWN_IOCPTHREAD, null);
                    }
                }
                // Wait here until all the threads are gone
                while (ThreadsInPool != 0) Thread.Sleep(100);
                unsafe
                {
                    // Close the IOCP Handle
                    IOCPEx.CloseHandle(GetHandle);
                }
            }
            catch
            {
            }
        }


        private void IOCPFunction()
        {
            UInt32 uiNumberOfBytes;
            Int32 iValue;
            try
            {
                while (true)
                {
                    unsafe
                    {
                        OVERLAPPED* pOv;
                        // Wait for an event
                        IOCPEx.GetQueuedCompletionStatus(GetHandle, &uiNumberOfBytes, (UInt32*)&iValue, &pOv, IOCPEx.INIFINITE);
                    }
                    // Decrement the number of events in queue
                    this.DecCurrentWorkInPool();
                    // Was this thread told to shutdown
                    if (iValue == IOCPEx.SHUTDOWN_IOCPTHREAD)
                    {
                        break;
                    }
                    // Increment the number of active threads
                    this.IncActThreadsInPool();
                    try
                    {
                        // Call the user function
                        this.UserFunction(iValue);
                    }
                    catch
                    {
                    }
                    // Get a lock, 
                    //TODO:move the lock to unluck method
                    Monitor.Enter(CriticalSection);
                    try
                    {
                        // If we have less than max threads currently in the pool
                        if (ThreadsInPool < MaxThreadsInPool)
                        {
                            // Should we add a new thread to the pool
                            if (ActiveThreadsInPool == ThreadsInPool)
                            {
                                if (IsDisposed == false)
                                {
                                    // Create a thread and start it
                                    ThreadStart tsThread = new ThreadStart(IOCPFunction);
                                    Thread thThread = new Thread(tsThread);
                                    thThread.Name = "IOCP " + thThread.GetHashCode();
                                    thThread.Start();
                                    // Increment the thread pool count
                                    IncCurrentThreadsInPool();
                                }
                            }
                        }
                    }
                    catch
                    {
                    }
                    // Relase the lock
                    Monitor.Exit(CriticalSection);
                    // Increment the number of active threads
                    DecActThreadsInPool();
                }
            }
            catch
            {
            }
            // Decrement the thread pool count
            DecCurrentThreadsInPool();
        }

        /// <summary> IOCP Worker Function that calls the specified user function </summary>
        /// <param name="iValue"> SimType: A value to be passed with the event </param>
        /// <exception cref = "Exception"> Unhandled Exception </exception>
        public void PostEvent(Int32 iValue)
        {
            try
            {
                // Only add work if we are not disposing
                if (IsDisposed == false)
                {
                    unsafe
                    {
                        // Post an event into the IOCP Thread Pool
                        IOCPEx.PostQueuedCompletionStatus(GetHandle, 4, (UInt32*)iValue, null);
                    }
                    // Increment the number of item of work
                    IncCurrentWorkInPool();
                    // Get a lock
                    Monitor.Enter(CriticalSection);
                    try
                    {
                        // If we have less than max threads currently in the pool
                        if (ThreadsInPool < MaxThreadsInPool)
                        {
                            // Should we add a new thread to the pool
                            if (ActiveThreadsInPool == ThreadsInPool)
                            {
                                if (IsDisposed == false)
                                {
                                    // Create a thread and start it
                                    ThreadStart tsThread = new ThreadStart(IOCPFunction);
                                    Thread thThread = new Thread(tsThread);
                                    thThread.Name = "IOCP " + thThread.GetHashCode();
                                    thThread.Start();
                                    // Increment the thread pool count
                                    IncCurrentThreadsInPool();
                                }
                            }
                        }
                    }
                    catch
                    {
                    }
                    // Release the lock
                    Monitor.Exit(CriticalSection);
                }
            }
            catch (Exception e)
            {
                throw e;
            }
            catch
            {
                throw new Exception("Unhandled Exception");
            }
        }
        /// <summary> IOCP Worker Function that calls the specified user function </summary>
        /// <exception cref = "Exception"> Unhandled Exception </exception>
        public void PostEvent()
        {
            try
            {
                // Only add work if we are not disposing
                if (IsDisposed == false)
                {
                    unsafe
                    {
                        // Post an event into the IOCP Thread Pool
                        IOCPEx.PostQueuedCompletionStatus(GetHandle, 0, null, null);
                    }
                    // Increment the number of item of work
                    IncCurrentWorkInPool();
                    // Get a lock
                    Monitor.Enter(CriticalSection);
                    try
                    {
                        // If we have less than max threads currently in the pool
                        if (ThreadsInPool < MaxThreadsInPool)
                        {
                            // Should we add a new thread to the pool
                            if (ActiveThreadsInPool == ThreadsInPool)
                            {
                                if (IsDisposed == false)
                                {
                                    // Create a thread and start it
                                    ThreadStart tsThread = new ThreadStart(IOCPFunction);
                                    Thread thThread = new Thread(tsThread);
                                    thThread.Name = "IOCP " + thThread.GetHashCode();
                                    thThread.Start();
                                    // Increment the thread pool count
                                    IncCurrentThreadsInPool();
                                }
                            }
                        }
                    }
                    catch
                    {
                    }
                    // Release the lock
                    Monitor.Exit(CriticalSection);
                }
            }
            catch (Exception e)
            {
                throw e;
            }
            catch
            {
                throw new Exception("Unhandled Exception");
            }
        }
    }
}